package tech.abing.spider.util;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.util.List;

import org.apache.log4j.Logger;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

import tech.abing.spider.constants.FinalConstants;

/**
 * 操作Redis队列的工具类
 * 
 * @Description: TODO
 * @author Ivan 2862099249@qq.com
 * @date 2015年8月22日 下午4:47:43
 * @version V1.0
 */
public class RedisUtil {

	private static Logger logger = Logger.getLogger(RedisUtil.class);
	
	private JedisPool pool = null;

	private byte[] queue;
	
	public static RedisUtil getInstance(){
		return new RedisUtil(SysUtil.getValue("redis.server"), 6379);
	}
	
	public String getQueue(){
		
		try {
			return new String(this.queue, FinalConstants.ENCODING_UTF8);
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}
		return null;
	}
	
	/**
	 * 给定JedisPoolConfig初始化一个队列
	 * @param host
	 * @param port
	 * @param config
	 * @param queue
	 */
	public RedisUtil(String host, int port, JedisPoolConfig config, String queue) {
		pool = new JedisPool(config, host, port, 15000);
		try {
			this.queue = queue.getBytes(FinalConstants.ENCODING_UTF8);
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * 基于默认配置初始化一个队列
	 * @param host
	 * @param port
	 * @param queue
	 */
	public RedisUtil(String host, int port, String queue) {
		
		this(host, port);

		try {
			this.queue = queue.getBytes(FinalConstants.ENCODING_UTF8);
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * 基于默认配置初始化连接池
	 * @param host
	 * @param port
	 * @param queue
	 */
	public RedisUtil(String host, int port) {
		JedisPoolConfig config = new JedisPoolConfig();

		// 连接耗尽时是否阻塞, false报异常,ture阻塞直到超时, 默认true
		config.setBlockWhenExhausted(true);

		// 设置的逐出策略类名, 默认DefaultEvictionPolicy(当连接超过最大空闲时间,或连接数超过最大空闲连接数)
		config.setEvictionPolicyClassName("org.apache.commons.pool2.impl.DefaultEvictionPolicy");

		// 是否启用pool的jmx管理功能, 默认true
		config.setJmxEnabled(true);

		// MBean ObjectName = new
		// ObjectName("org.apache.commons.pool2:type=GenericObjectPool,name=" +
		// "pool" + i); 默 认为"pool", JMX不熟,具体不知道是干啥的...默认就好.
		config.setJmxNamePrefix("pool");

		// 是否启用后进先出, 默认true（针对的是连接）
		config.setLifo(true);

		// 最大空闲连接数, 默认8个
		config.setMaxIdle(30);

		// 最大连接数, 默认8个
		config.setMaxTotal(100);

		// 获取连接时的最大等待毫秒数(如果设置为阻塞时BlockWhenExhausted),如果超时就抛异常, 小于零:阻塞不确定的时间,
		// 默认-1
		config.setMaxWaitMillis(10000);

		// 逐出连接的最小空闲时间 默认1800000毫秒(30分钟)
		config.setMinEvictableIdleTimeMillis(1800000);

		// 最小空闲连接数, 默认0
		config.setMinIdle(10);

		// 每次逐出检查时 逐出的最大数目 如果为负数就是 : 1/abs(n), 默认3
		config.setNumTestsPerEvictionRun(3);

		// 对象空闲多久后逐出, 当空闲时间>该值 且 空闲连接>最大空闲数
		// 时直接逐出,不再根据MinEvictableIdleTimeMillis判断 (默认逐出策略)
		config.setSoftMinEvictableIdleTimeMillis(1800000);

		// 在获取连接的时候检查有效性, 默认false
		config.setTestOnBorrow(true);

		// 在空闲时检查有效性, 默认false
		config.setTestWhileIdle(true);

		// 逐出扫描的时间间隔(毫秒) 如果为负数,则不运行逐出线程, 默认-1
		config.setTimeBetweenEvictionRunsMillis(20);

		pool = new JedisPool(config, host, port, 15000);
		
	}
	
	/**
	 * 入队到初始化队列队尾中
	 * @param e
	 * @return
	 */
	public boolean lpush(final Object e) {
		
		return this.lpush(e, this.getQueue());
		
	}
	
	/**
	 * 入队到指定队列队尾中
	 * @param e
	 * @param disQueue
	 * @return
	 */
	public boolean lpush(final Object e, final String disQueue) {
		
		new OperationJedis<Void>() {

			@Override
			protected Void operation(Jedis jedis) throws Exception {
				byte[] data = objectToByte(e);
				jedis.lpush(disQueue.getBytes(FinalConstants.ENCODING_UTF8), data);
				return null;
			}
		}.exe();
		return true;
	}
	
	public boolean lpush(final String str, final String disQueue) {
		
		new OperationJedis<Void>() {

			@Override
			protected Void operation(Jedis jedis) throws Exception {
				jedis.lpush(disQueue, str);
				return null;
			}
		}.exe();
		return true;
	}
	
	/**
	 * 入队到初始化队列队头中
	 * @param e
	 * @return
	 */
	public boolean rpush(final Object e) {
		
		return this.rpush(e, this.getQueue());
		
	}
	
	/**
	 * 入队到指定队列队头中
	 * @param e
	 * @param disQueue
	 * @return
	 */
	public boolean rpush(final Object e, final String disQueue) {
		
		new OperationJedis<Void>() {

			@Override
			protected Void operation(Jedis jedis) throws Exception {
				byte[] data = objectToByte(e);
				jedis.rpush(disQueue.getBytes(FinalConstants.ENCODING_UTF8), data);
				return null;
			}
		}.exe();
		return true;
	}
	
	/**
	 * 从初始化队列中出队
	 * @return
	 * @throws InterruptedException
	 */
	public Object bpop() throws InterruptedException {

		return this.bpop(getQueue());
	}
	
	/**
	 * 普通出队，比如:字符串类型
	 * @param disQueue
	 * @return
	 * @throws InterruptedException
	 */
	public String brpop(final String disQueue) throws InterruptedException {
		final List<String> dataList = new OperationJedis<List<String>>() {

			@Override
			protected List<String> operation(Jedis jedis) throws Exception {
				
				List<String> list = jedis.brpop(0, disQueue);//阻塞式出队
				return list;
			}
		}.exe();
		
		String data = null;
		if(dataList != null && dataList.size() == 2){
			//获取key
			String key = dataList.get(0);
			
			data = dataList.get(1);
			
			logger.info("["+key+"]出队, 出队元素["+data+"]");
		}
		
		return data;
	}
	
	public Object rpopObj(final String disQueue) {
		Object obj = null;

		byte[] data = new OperationJedis<byte[]>() {

			@Override
			protected byte[] operation(Jedis jedis) throws Exception {
				return jedis.rpop(disQueue.getBytes(FinalConstants.ENCODING_UTF8));
			}
		}.exe();

		obj = byteToObject(data);

		return obj;
	}
	
	public String rpopStr(final String disQueue) {
		final String data = new OperationJedis<String>() {

			@Override
			protected String operation(Jedis jedis) throws Exception {
				
				return jedis.rpop(disQueue);
			}
		}.exe();
		if(StringUtils.isNotBlank(data)){
			logger.info("["+disQueue+"]出队, 出队元素["+data+"]");
		}
		return data;
	}
	
	/**
	 * 从队列组中出队，按优先顺序，别高的先出队
	 * @param disQueues
	 * @return
	 * @throws InterruptedException
	 */
	public Object bpop(final String... disQueues) throws InterruptedException {

		if(disQueues == null || disQueues.length == 0){
			throw new RuntimeException("必须给定出队队列名！");
		}
		
		Object obj = null;
		
		final List<byte[]> dataList = new OperationJedis<List<byte[]>>() {

			@Override
			protected List<byte[]> operation(Jedis jedis) throws Exception {
				
				byte[][] dqByte = new byte[disQueues.length][];
				
				for(int i = 0; i < disQueues.length; i++){
					dqByte[i] = disQueues[i].getBytes(FinalConstants.ENCODING_UTF8);
				}
				
				//阻塞式出队，返回一个含有两个元素的列表，第一个元素是被弹出元素所属的 key ，第二个元素是被弹出元素的值。
				List<byte[]> list = jedis.brpop(0, dqByte);//0表示一直等待
				
				return list;
			}
		}.exe();

		if(dataList != null && dataList.size() == 2){
			//获取key
			byte[] key = dataList.get(0);
			
			byte[] data = dataList.get(1);
			
			String keyStr = "";
			try {
				//不能byteToObject转换，否则java.io.StreamCorruptedException: invalid stream header: 70726F64异常
				keyStr = new String(key, FinalConstants.ENCODING_UTF8);
			} catch (UnsupportedEncodingException e) {
				e.printStackTrace();
			}
			
			obj = byteToObject(data);
			
			logger.info("["+keyStr+"]出队, 出队元素["+obj+"]");
		}
		
		return obj;
	}
	
	/**
	 * 返回队列中的元素个数。
	 */
	public int size() {
		return this.size(this.getQueue());
	}
	
	/**
	 * 返回指定队列的元素个数
	 * @param disQueues
	 * @return
	 */
	public int size(final String... disQueues) {
		long size = new OperationJedis<Long>() {

			@Override
			protected Long operation(Jedis jedis) throws Exception {
				
				long len = 0;
				for(int i = 0; i < disQueues.length; i++){
					len += jedis.llen(disQueues[i].getBytes(FinalConstants.ENCODING_UTF8));
				}
				return len;
			}
		}.exe();
		return (int) size;
	}
	
	/**
	 * Sorted Sets入队,zadd如果某个 member 已经是有序集的成员，那么更新这个 member 的 score 值，并通过重新插入这个 member 元素，
	 * 来保证该 member 在正确的位置上
	 * @param e
	 * @param scoreMembers
	 * @param disQueue
	 * @return
	 */
	public boolean zadd(final Object e, final double scoreMembers, final String disQueue){
		
		new OperationJedis<Void>() {

			@Override
			protected Void operation(Jedis jedis) throws Exception {
				
				jedis.zadd(disQueue.getBytes(FinalConstants.ENCODING_UTF8), scoreMembers, objectToByte(e));
				
				return null;
			}
		}.exe();
		
		return true;
	}
	
	/**
	 * 用Set集合来判断任务是否存在
	 * @param e
	 * @param disQueue
	 * @return
	 */
	public boolean exist(final Object e, final String disSet){
		
		long num = new OperationJedis<Long>() {

			@Override
			protected Long operation(Jedis jedis) throws Exception {
				
				return jedis.sadd(disSet.getBytes(FinalConstants.ENCODING_UTF8), objectToByte(e));
			}
		}.exe();
		
		return num == 0;
	}
	
	/**
	 * 添加成员到SET集合中
	 * @param member
	 * @param disSet
	 * @return
	 */
	public boolean sadd(final String member, final String disSet){
		
		long num = new OperationJedis<Long>() {

			@Override
			protected Long operation(Jedis jedis) throws Exception {
				
				return jedis.sadd(disSet, member);
			}
		}.exe();
		
		return num == 0;
	}
	
	/**
	 * 销毁Redis连接池
	 */
	public void destroy(){
		this.pool.destroy();
	}
	
	/**
	 * 抽取一个操作redis的回调抽象类
	 * 
	 * @Description: TODO
	 * @author Ivan 2862099249@qq.com
	 * @date 2015年8月22日 下午4:53:37 
	 * @version V1.0
	 */
	private abstract class OperationJedis<E> {

		protected abstract E operation(Jedis jedis) throws Exception;

		public final E exe() {
			Jedis jedis = null;
			E result = null;
			try {
				jedis = pool.getResource();
				result = operation(jedis);
			} catch (Exception e) {
				logger.error("获取redis连接异常", e);
				if (jedis != null) {
					pool.returnBrokenResource(jedis);
					jedis = null;
				}
			} finally {
				if (jedis != null) {
					pool.returnResource(jedis);
				}
			}
			return result;
		}
	}
	
	
	/**
	 * 对象转换为字节数组
	 * @param obj
	 * @return
	 */
	public byte[] objectToByte(Object obj) {
		if (obj == null) {
			return null;
		}
		byte[] bytes = null;
		ByteArrayOutputStream bo = new ByteArrayOutputStream();
		ObjectOutputStream oo = null;
		try {
			oo = new ObjectOutputStream(bo);
			oo.writeObject(obj);
			bytes = bo.toByteArray();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (oo != null) {
				try {
					oo.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			if (bo != null) {
				try {
					bo.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		return bytes;
	}

	/**
	 * 字节数组转换为对象
	 * @param bytes
	 * @return
	 */
	public Object byteToObject(byte[] bytes) {
		if (bytes == null) {
			return null;
		}
		Serializable obj = null;
		ByteArrayInputStream bi = new ByteArrayInputStream(bytes);
		ObjectInputStream oi = null;
		try {
			oi = new ObjectInputStream(bi);
			obj = (Serializable) oi.readObject();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (oi != null) {
				try {
					oi.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			if (bi != null) {
				try {
					bi.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		return obj;
	}
	

	public static void main(String[] args) {

		RedisUtil redisUtil = new RedisUtil("121.40.163.151", 6379);
		
//		ProductID p1 = new ProductID("1", "1", "1", "1", 1);
//		ProductID p2 = new ProductID("2", "2", "2", "2", 2);
//		ProductID p3 = new ProductID("3", "3", "3", "3", 3);
		
		redisUtil.lpush("aaa", "high_level");
		redisUtil.lpush("bbb", "middle_level");
		redisUtil.lpush("ccc", "low_level");
		
		redisUtil.lpush("111", "high_level");
		redisUtil.lpush("222", "middle_level");
		redisUtil.lpush("333", "low_level");
		
		redisUtil.lpush("dfdsfd", "high_level");
		redisUtil.lpush("bbsafb", "middle_level");
		redisUtil.lpush("ccsdfc", "low_level");
		
		redisUtil.lpush("11dfd1", "low_level");
		redisUtil.lpush("22dfd2", "middle_level");
		redisUtil.lpush("dff", "low_level");
		
		redisUtil.lpush("eeeaaa", "low_level");
		redisUtil.lpush("bberswerb", "middle_level");
		redisUtil.lpush("cccsdffdf", "low_level");
		
		redisUtil.lpush("sdfsdferw", "low_level");
		redisUtil.lpush("22ddddd2", "low_level");
		redisUtil.lpush("ddddddddd", "low_level");
		
		System.out.println(redisUtil.size("high_level", "middle_level", "low_level"));
		
//		redisUtil.lpush(p1);
//		redisUtil.lpush(p2);
//		redisUtil.lpush(p3);
		
		try {
			while(true){
				redisUtil.bpop("high_level", "middle_level", "low_level");
			}
			
		} catch (InterruptedException e) {
			logger.error("阻塞队列出队中断异常", e);
		}
	}

}
